package com.sardine.base.threads.master;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * 北京京东世纪贸易有限公司<br>
 * Created with IntelliJ IDEA.
 * Date:2015/10/1
 * Time:17:14
 * 多线程
 * @author daifeng<br>
 *         daifeng@jd.com
 * @since JDK 1.6 <br>
 * Copyright (c) 2015 JD System Incorporated All Rights Reserved.
 */
public class Master {
    private static final Logger logger = LoggerFactory.getLogger(Master.class);
    //一个同步辅助类
    protected CountDownLatch downLatch;
    //任务队列
    protected final Queue<Object> workerQueue=new ConcurrentLinkedQueue<Object>();
    //worker进程队列
    protected final ExecutorService executorService;
    //子任务处理结果集
    protected final Map<String,Object> resultMap=new ConcurrentHashMap<String,Object>();
    protected final MWorker mWorker;
    protected final int countWorker;

    public Master(MWorker mWorker,int countWorker,ExecutorService executorService) {
        this.mWorker=mWorker;
        this.countWorker=countWorker;
        this.executorService=executorService;
    }
    /**
     * 提交一个任务
     * @param job
     */
    public void submit(Object job){
        workerQueue.add(job);
    }

    /**
     * 提交一个任务
     * @param
     */
    public void submitCollection(Collection<?> c){
        workerQueue.addAll(c);
    }

    /**
     * 是否结束
     * @return
     */
    public boolean isComplete(){
       return downLatch.getCount()==0L;
    }

    /**
     * 返回结果集 ---该方法会阻塞
     * @return
     */
    public Map<String,Object> getResultMap(){
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            logger.error("多线程错误：CountDownLatch await error",e);
        }
        return resultMap;
    }

    /**
     * 开始执行所有的worker进程。进行处理；
     */
    public void execute(){
        this.downLatch = new CountDownLatch(workerQueue.size());
        mWorker.setDownLatch(downLatch);
        mWorker.setResultMap(resultMap);
        mWorker.setWorkerQueue(workerQueue);
        for(int i=0;i<countWorker;i++){
            executorService.execute(mWorker);
        }
    }


}
